package com.atguigu.flink.state;

import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Smexy on 2022/11/23
 *
 *      有状态的计算：   每一条数据之间，需要交互一些数据。
 *                          单词统计
 *
 *      无状态的计算:   每一条数据之间，各算各的。
 *
 *
 *      状态： 算子执行的XXXFunction中声明的 非只读的 属性
 *
 *    ---------------------
 *          保证，程序出现故障时，状态不丢失！
 *
 *    -------------
 *      Task的重启:
 *                      flink如果检测到某个Task出现了异常，会重新将Task发送到一个新的Slot，重试。 称为Task的 failover
 *
 *                      flink尝试去恢复task，被 NoRestartBackoffTimeStrategy 抑制了
 *                      org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
 *      Job的重启：
 *                      cacel 掉 job后，重新run
 *
 *     -----------------
 *          Checkpoint: 备份，存档
 */
public class Demo1_State
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);


        env.setParallelism(2);
        //设置Task的失败重启策略    当Task失败时，尝试重启3次，每次重启会间隔3s
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(3)));

        //开启checkpoint，每间隔3000ms，做一次ck  Task无限重启
        env.enableCheckpointing(500);
        //存到文件系统
        env.getCheckpointConfig().setCheckpointStorage("file:///e://ck");
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:9820/ck");

        //设置状态后端，默认
        env.setStateBackend(new HashMapStateBackend());

        //使用rocksdb,额外引入依赖
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        //读取数据
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        //把每次读到的数据存入到一个List集合中
        ds.flatMap(new MyFlatMap())
          .addSink(new SinkFunction<String>()
          {
              @Override
              public void invoke(String value, Context context) throws Exception {
                  if (value.contains("x")){
                      throw new RuntimeException("出异常了...");
                  }
                  System.out.println(value);
              }
          });

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    private static class  MyFlatMap implements  FlatMapFunction<String,String>, CheckpointedFunction
    {

        //不要声明普通的类型的 属性，而应该用Flink提供的 ManagedState(属性，自动备份)
       // private List<String> strs=new ArrayList<>();
        // add: 加一个
        // addAll: 加一个List
        // clear: 清空
        // update ： 等价于 先 clear再addAll
        // get：  获取里面的元素
        private ListState<String> state;

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {

            //strs.add(value);
            state.add(value);

            out.collect(MyUtil.parseList(state.get()).toString());

        }

        //快照状态，备份  周期型执行
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyFlatMap.snapshotState");
        }

        //初始化状态  恢复  只在Task'启动时执行一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyFlatMap.initializeState");

            //声明状态, 从备份中读取状态
            state = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("list", String.class));

        }
    }
}
